"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 行存压缩
Case Name   : 分区表非压缩表，修改为压缩表，gs_probackup备份恢复数据后，使用gs_basebackup备份
Create At   : 2023/8/10
Owner       : opentestcase013
Description :
    1、建分区表（非压缩表），并插入数据
    2、修改分区表为压缩表
    3、查看表占用磁盘大小、数据量等
    4、gs_probackup备份
    5、恢复
    6、查看恢复成功
    7、使用gs_basebackup备份
    8、使用备份目录启动数据库
    9、查询测试表是否存在，查看表占用磁盘大小、数据量等，与备份前对比
    10、清理环境
Expect      :
    1、成功
    2、成功
    3、成功
    4、成功
    5、成功
    6、成功
    7、成功
    8、成功
    9、与备份前一致
    10、成功
History     :
"""

import os
import unittest
import re

from yat.test import Node
from yat.test import macro

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger

Pri_U_SH = CommonSH('PrimaryDbUser')
Node_Num = Pri_U_SH.get_node_num()


@unittest.skipIf(3 > Node_Num, '不满足一主两备环境跳过')
class RowCompressTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.std1_sh = CommonSH('Standby1DbUser')
        self.std2_sh = CommonSH('Standby2DbUser')
        self.pri_user = Node('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.tb_name = 't_alter_rowcompress_0066'
        self.backup_dir = ['f_aler_rowcompress_0066_01',
                           'f_aler_rowcompress_0066_02']
        self.proback_path = os.path.join(os.path.dirname(
            macro.DB_INSTANCE_PATH), self.backup_dir[0])
        self.baseback_path = os.path.join(os.path.dirname(
            macro.DB_INSTANCE_PATH), self.backup_dir[1])
        self.alter_compress_param = 'compresstype=2, ' \
                                    'compress_chunk_size=1024, ' \
                                    'compress_prealloc_chunks=6, ' \
                                    'compress_byte_convert=true, ' \
                                    'compress_diff_convert=true, ' \
                                    'compress_level=6'

    def test_row_compress(self):
        text = '-----step1:建分区表（非压缩表），并插入数据;expect:成功-----'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name} cascade;
                create table {self.tb_name}
                (prod_id number(6),
                 cust_id number,
                 time_id date,
                 channel_id char(1),
                 promo_id number(6),
                 quantity_sold number(3),
                 amount_sold number(10,2)
                )
                partition by range (time_id)
                interval('1 day')
                ( partition p1 values less than ('2023-02-01 00:00:00'),
                  partition p2 values less than ('2023-02-02 00:00:00')
                );
                insert into {self.tb_name} values(generate_series(0, 1000), 
                lpad(floor(random() * 1000)::text, 2, '0'), 
                '2023-01-10 00:00:00', 'a', 1, 1, 1);
                insert into {self.tb_name} values(generate_series(0, 1000), 
                lpad(floor(random() * 1000)::text, 2, '0'), 
                '2023-02-01 00:00:00', 'a', 2, 2, 2);
                '''
        sql_res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn('INSERT 0 1001', sql_res, '执行失败：' + text)
        self.assertEqual(sql_res.count('INSERT 0 1001'), 2, '执行失败：' + text)

        text = '-----step2:修改分区表为压缩表;expect:成功-----'
        self.log.info(text)
        sql = f'alter table {self.tb_name} set ' \
              f'({self.alter_compress_param}); checkpoint;' \
              f'\\d+ {self.tb_name}'
        sql_res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.ALTER_TABLE_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CHECKPOINT_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(f'{self.alter_compress_param}', sql_res,
                      '执行失败：' + text)

        text = '-----step3:查看表占用磁盘大小、数据量等;expect:成功-----'
        self.log.info(text)
        sql = f"select pg_size_pretty(pg_total_relation_size(" \
              f"'{self.tb_name}')) as table_size;"
        res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(res)
        self.assertNotIn('ERROR', res, '执行失败：' + text)

        sql = f'select checksum({self.tb_name}::text) from {self.tb_name};'
        before_size = Pri_U_SH.execut_db_sql(sql)
        self.log.info(before_size)
        self.assertNotIn('ERROR', before_size, '执行失败：' + text)

        text = '-----step4:gs_probackup备份;expect:成功-----'
        self.log.info(text)
        self.log.info('---新建目录，进行初始化---')
        cmd = f"rm -rf {self.proback_path};mkdir -p {self.proback_path};" \
              f"chmod 700 {self.proback_path};" \
              f"ls {os.path.dirname(macro.DB_INSTANCE_PATH)} | " \
              f"grep '{self.backup_dir[0]}';"
        self.log.info(cmd)
        res = self.common.get_sh_result(self.pri_user, cmd)
        self.log.info(res)
        self.assertIn(f'{self.backup_dir[0]}', res, '执行失败：' + text)

        cmd = f'source {macro.DB_ENV_PATH};' \
              f'gs_probackup init -B {self.proback_path}'
        self.log.info(cmd)
        res = self.common.get_sh_result(self.pri_user, cmd)
        self.assertIn('successfully inited', res, '执行失败：' + text)

        self.log.info('---在备份路径内初始化一个新的备份实例---')
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gs_probackup add-instance -B {self.proback_path} " \
              f"-D {macro.DB_INSTANCE_PATH} --instance=pro1"
        self.log.info(cmd)
        res = self.common.get_sh_result(self.pri_user, cmd)
        self.assertIn('successfully inited', res, '执行失败：' + text)

        self.log.info('---进行全量备份---')
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gs_probackup backup -B {self.proback_path} -b full --stream " \
              f"--instance=pro1 --skip-block-validation " \
              f"--no-validate -p {self.pri_user.db_port} " \
              f"-d {self.pri_user.db_name}"
        self.log.info(cmd)
        res = self.common.get_sh_result(self.pri_user, cmd)
        re.search('Backup * completed', res)

        self.log.info('---查看备份情况---')
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gs_probackup show -B {self.proback_path} --instance=pro1"
        self.log.info(cmd)
        res = self.common.get_sh_result(self.pri_user, cmd)
        self.assertIn('DONE', res, '执行失败：' + text)

        self.log.info('---删除表，停止数据库---')
        sql = f"drop table if exists {self.tb_name};"
        self.log.info(sql)
        res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(res)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, res,
                      '执行失败：' + text)

        stop_res = Pri_U_SH.stop_db_instance()
        self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG, stop_res,
                      '执行失败：' + text)

        text = '-----step5:恢复;expect:成功-----'
        self.log.info(text)
        self.log.info('---执行restore，并重启数据库---')
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gs_probackup restore -B {self.proback_path} --instance=pro1 " \
              f"--incremental-mode=checksum -D {macro.DB_INSTANCE_PATH};"
        self.log.info(cmd)
        res = self.common.get_sh_result(self.pri_user, cmd)
        re.search('Restore of backup * completed', res)

        start_res = Pri_U_SH.start_db_instance()
        self.assertIn('server started', start_res, '启动失败')

        status = Pri_U_SH.get_db_cluster_status(param='status')
        self.log.info(status)
        if status is False:
            build_res1 = self.std1_sh.build_standby(mode='')
            self.assertIn(self.constant.BUILD_SUCCESS_MSG, build_res1)
            build_res2 = self.std2_sh.build_standby(mode='')
            self.assertIn(self.constant.BUILD_SUCCESS_MSG, build_res2)
            status = Pri_U_SH.get_db_cluster_status(param='status')
            self.log.info(status)

        text = '-----step6:查看恢复成功;expect:成功-----'
        self.log.info(text)
        sql = f"select pg_size_pretty(pg_total_relation_size(" \
              f"'{self.tb_name}')) as table_size;"
        res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(res)
        self.assertNotIn('ERROR', res, '执行失败：' + text)

        sql = f'select checksum({self.tb_name}::text) from {self.tb_name};'
        after_size = Pri_U_SH.execut_db_sql(sql)
        self.log.info(after_size)
        self.assertNotIn('ERROR', before_size, '执行失败：' + text)
        self.assertEqual(before_size, after_size, '执行失败：' + text)

        text = '-----step7:使用gs_basebackup备份;expect:成功-----'
        self.log.info(text)
        self.log.info('---建目录---')
        cmd = f"rm -rf {self.baseback_path};mkdir -p {self.baseback_path};" \
              f"chmod 700 {self.baseback_path};" \
              f"ls {os.path.dirname(macro.DB_INSTANCE_PATH)} | " \
              f"grep '{self.backup_dir[1]}';"
        self.log.info(cmd)
        res = self.common.get_sh_result(self.pri_user, cmd)
        self.assertIn(f'{self.backup_dir[1]}', res, '执行失败：' + text)

        self.log.info('---执行备份---')
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gs_basebackup -D {self.baseback_path} -Fp -X stream " \
              f"-p {self.pri_user.db_port} -l gauss_042.bak " \
              f"-U {self.pri_user.ssh_user} -w"
        self.log.info(cmd)
        res = self.common.get_sh_result(self.pri_user, cmd)
        self.assertIn('base backup successfully', res, '执行失败：' + text)

        self.log.info('---停止数据库---')
        stop_res = Pri_U_SH.stop_db_instance()
        self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG, stop_res,
                      '执行失败：' + text)

        self.log.info('-----step8:使用备份目录启动数据库;expect:成功-----')
        start_res = Pri_U_SH.start_db_instance(dn_path=self.baseback_path)
        self.assertIn(self.constant.RESTART_SUCCESS_MSG, start_res,
                      '执行失败：' + text)
        res1 = self.std1_sh.build_standby(mode='')
        self.assertIn(self.constant.BUILD_SUCCESS_MSG, res1, '备1build失败')
        res2 = self.std2_sh.build_standby(mode='')
        self.assertIn(self.constant.BUILD_SUCCESS_MSG, res2, '备2build失败')

        cmd = f"source {macro.DB_ENV_PATH};" \
              f"gs_ctl query -D {self.baseback_path}"
        self.log.info(cmd)
        status = self.common.get_sh_result(self.pri_user, cmd)
        self.assertTrue('Normal' or 'Degraded' in status, '数据库状态异常')

        text = '-----step9:再次查看表占用磁盘大小、数据量等，与备份前对比;' \
               'expect:查看数据与备份前一致-----'
        self.log.info(text)
        sql = f"select pg_size_pretty(pg_total_relation_size(" \
              f"'{self.tb_name}')) as table_size;"
        res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(res)
        self.assertNotIn('ERROR', res, '执行失败：' + text)

        sql = f'select checksum({self.tb_name}::text) from {self.tb_name};'
        after_size = Pri_U_SH.execut_db_sql(sql)
        self.log.info(after_size)
        self.assertNotIn('ERROR', before_size, '执行失败：' + text)
        self.assertEqual(before_size, after_size, '执行失败：' + text)

    def tearDown(self):
        text = '-----step10:清理环境; expect:成功-----'
        self.log.info(text)
        stop_res = Pri_U_SH.stop_db_instance(dn_path=self.baseback_path)
        self.log.info(stop_res)
        start_res = Pri_U_SH.start_db_instance()
        self.log.info(start_res)

        cmd = f'rm -rf {self.proback_path};' \
              f'if [ -d {self.proback_path} ]; ' \
              f'then echo "exists"; else echo "not exists"; fi'
        rm_res1 = self.common.get_sh_result(self.pri_user, cmd)
        self.log.info(rm_res1)

        cmd = f'rm -rf {self.baseback_path};' \
              f'if [ -d {self.baseback_path} ]; ' \
              f'then echo "exists"; else echo "not exists"; fi'
        rm_res2 = self.common.get_sh_result(self.pri_user, cmd)
        self.log.info(rm_res2)

        sql = f'drop table if exists {self.tb_name};'
        drop_res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(drop_res)

        status = Pri_U_SH.get_db_cluster_status(param='status')
        self.log.info(status)
        if status is False:
            build_res1 = self.std1_sh.build_standby(mode='')
            self.assertIn(self.constant.BUILD_SUCCESS_MSG, build_res1)
            build_res2 = self.std2_sh.build_standby(mode='')
            self.assertIn(self.constant.BUILD_SUCCESS_MSG, build_res2)
            status = Pri_U_SH.get_db_cluster_status(param='status')
            self.log.info(status)

        self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG, stop_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.RESTART_SUCCESS_MSG, start_res,
                      '执行失败：' + text)
        self.assertEqual('not exists', rm_res1, '文件删除失败')
        self.assertEqual('not exists', rm_res2, '文件删除失败')
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_res,
                      '执行失败：' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
